package org.finesys.common.redis.core.mq.pubsub;

import java.lang.reflect.Type;
import java.util.List;

import org.finesys.commom.jackson.util.JacksonUtil;
import org.finesys.common.redis.core.mq.client.RedisMQTemplate;
import org.finesys.common.redis.core.mq.interceptor.RedisMessageInterceptor;
import org.finesys.common.redis.core.mq.message.AbstractRedisMessage;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

import cn.hutool.core.util.TypeUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;

/**
 * 订阅消息
 */
@Getter
@Setter
public abstract class AbstractPubSubMessageListener<T extends AbstractPubSubMessage> implements MessageListener {

	private final Class<T> messageType;

	private final String topic;

	private RedisMQTemplate redisMQTemplate;

	@SneakyThrows
	protected AbstractPubSubMessageListener() {
		this.messageType = getMessageClass();
		this.topic = messageType.newInstance().getTopic();
	}

	private Class<T> getMessageClass() {
		Type type = TypeUtil.getTypeArgument(getClass(), 0);
		if (type == null) {
			throw new IllegalStateException(String.format("类型(%s)需要设置消息类型", getClass().getName()));
		}
		return (Class<T>) type;
	}

	/**
	 * 处理消息
	 */
	@Override
	public final void onMessage(Message message, byte[] pattern) {
		T messageObj = JacksonUtil.parseObject(message.getBody(), messageType);
		try {
			consumeMessageBefore(messageObj);
			this.onMessage(messageObj);
		}
		finally {
			consumeMessageAfter(messageObj);
		}
	}

	/**
	 * 处理消息
	 */
	public abstract void onMessage(T message);

	private void consumeMessageAfter(AbstractRedisMessage message) {
		List<RedisMessageInterceptor> interceptors = redisMQTemplate.getInterceptors();
		for (int i = interceptors.size() - 1; i >= 0; i--) {
			interceptors.get(i).consumeMessageAfter(message);
		}
	}

	private void consumeMessageBefore(AbstractRedisMessage message) {
		List<RedisMessageInterceptor> interceptors = redisMQTemplate.getInterceptors();
		interceptors.forEach(interceptor -> interceptor.consumeMessageAfter(message));
	}

}
